#include "Receiver.h"
#include "CpuAffinity.h"

#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <iostream>
#include <netinet/in.h>
#include "RateMonitor.h"

Receiver::Receiver(const char *ip, uint16_t port,
                   LockFreeRingBuffer<OrderPacket> &ringBuffer,
                   MemoryPool &memoryPool,
                   int cpu_id)
    : ringBuffer_(ringBuffer), memoryPool_(memoryPool), cpu_id_(cpu_id)
{
    sockfd_ = socket(AF_INET, SOCK_DGRAM, 0);
    if (sockfd_ < 0)
    {
        perror("socket");
        exit(1);
    }

    // 允许地址复用
    int reuse = 1;
    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0)
    {
        perror("setsockopt SO_REUSEADDR");
        exit(1);
    }

    // 尝试开启 Busy Poll （需要root权限且内核支持）
    int busy_poll = 50; // 忙轮询时间 (微秒)
    if (setsockopt(sockfd_, SOL_SOCKET, SO_BUSY_POLL, &busy_poll, sizeof(busy_poll)) < 0)
    {
        perror("setsockopt SO_BUSY_POLL");
        // 不 exit，只警告，因为 busy_poll不是必须
    }

    // // 开启硬件时间戳优化
    // int timestamp_flags = SOF_TIMESTAMPING_RX_HARDWARE |
    // SOF_TIMESTAMPING_SOFTWARE |
    // SOF_TIMESTAMPING_RAW_HARDWARE;

    // setsockopt(sockfd_, SOL_SOCKET, SO_TIMESTAMPING, &timestamp_flags, sizeof(timestamp_flags));

    // 设置非阻塞
    int flags = fcntl(sockfd_, F_GETFL, 0);
    if (flags < 0)
    {
        perror("fcntl F_GETFL");
        exit(1);
    }
    if (fcntl(sockfd_, F_SETFL, flags | O_NONBLOCK) < 0)
    {
        perror("fcntl F_SETFL O_NONBLOCK");
        exit(1);
    }

    // 绑定地址
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = inet_addr(ip);

    if (bind(sockfd_, (sockaddr *)&addr, sizeof(addr)) < 0)
    {
        perror("bind");
        exit(1);
    }
}

Receiver::~Receiver()
{
    stop();
    if (sockfd_ >= 0)
    {
        close(sockfd_);
    }
}

void Receiver::start()
{
    running_ = true;
    thread_ = std::thread(&Receiver::run, this);
}

void Receiver::stop()
{
    running_ = false;
    if (thread_.joinable())
    {
        thread_.join();
    }
}

void Receiver::run()
{
    CpuAffinity::bindThreadToCpu(cpu_id_);

    constexpr size_t BUFFER_SIZE = 2048;
    char buffer[BUFFER_SIZE];

    while (running_)
    {
        sockaddr_in src_addr{};
        socklen_t addrlen = sizeof(src_addr);

        ssize_t len = recvfrom(sockfd_, buffer, sizeof(buffer), 0, (sockaddr *)&src_addr, &addrlen);
        if (len > 0)
        {
            handlePacket(buffer, len);
        }
        else
        {
            // 忙轮询，不sleep
            continue;
        }
    }
}

void Receiver::handlePacket(const char *data, size_t len)
{
    // rateMonitor_->incRecv();

    // 拷贝数据
    size_t offset = 0;
    while (len >= sizeof(OrderPacket))
    {
        // 从内存池分配
        size_t i = 0;
        OrderPacket *packet = nullptr;
        for (size_t i = 0; i < memoryPool_.getBlockSize(); i++)
        {
            packet = static_cast<OrderPacket *>(memoryPool_.allocate(i));
            if (packet == nullptr)
            {
                continue;
            }
        }

        if (!packet)
        {
            std::cerr << "MemoryPool exhausted! Drop packet!" << std::endl;
            packet = new OrderPacket();
            if (!packet)
            {
                return; // 内存不足，退出
            }
        }

        memcpy(packet, data + offset, sizeof(OrderPacket));
        printf("recv packet: %s %s %s %d %d %llu %u %llu\n", packet->accountId, packet->orderId, packet->symbol, (int)packet->isBuy, packet->orderType, packet->price, packet->quantity, packet->timestamp);
        // 推入无锁环形缓冲区
        while (!ringBuffer_.push(*packet))
        {
            // 若缓冲区满，忙等待
            continue;
        }
        offset += sizeof(OrderPacket);
        len -= sizeof(OrderPacket);
    }
}
